package com.atguigu.actual.edu0417.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @author: techyang
 * @since: 2023-10-19 15:30
 * @description:
 **/
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {
    private MapStateDescriptor<String,TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    // private BroadcastState<String,TableProcess> dimConfigState;

    private Map<String,TableProcess> configMap = new HashMap<String,TableProcess>();

    @Override
    public void open(Configuration parameters) throws Exception {
        // 如果广播流数据后到,先从map中读取配置信息
        // 通过JDBC获取配置信息
        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop101:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        PreparedStatement ps = conn.prepareStatement("select * from table_process_dim");
        ResultSet resultSet = ps.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();
        while (resultSet.next()) {
            JSONObject jsonObj = new JSONObject();
            int columnCount = metaData.getColumnCount();
            for (int i = 1; i <= columnCount; i++) {
                jsonObj.put(metaData.getColumnName(i),resultSet.getString(i));
            }
            TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }
        resultSet.close();
        ps.close();
        conn.close();
    }

    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        String tableName = jsonObj.getString("table");

        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        // 状态中拿不到配置信息,看map中是否能拿到配置信息
        TableProcess tableProcess = broadcastState.get(tableName) == null ? configMap.get(tableName) : broadcastState.get(tableName);
        if (Objects.nonNull(tableProcess)) {
            // 是维度表
            // 取data节点数据
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            // 过滤掉没必要的字段
            filterFields(dataJsonObj,tableProcess);
            // 补充信息
            // 数据操作类型
            dataJsonObj.put("type",jsonObj.getString("type"));
            // 时间戳字段
            //dataJsonObj.put("ts",jsonObj.getLong("ts") * 1000);
            // 配置信息
            dataJsonObj.put("tableProcess",tableProcess);
            out.collect(dataJsonObj);
        }
    }

    /**
     * 过滤不必要的字段
     * @param dataJsonObj
     * @param tableProcess
     */
    private void filterFields(JSONObject dataJsonObj, TableProcess tableProcess) {
        String[] sinkColumns = tableProcess.getSinkColumns().split(",");
        dataJsonObj.entrySet().removeIf(s-> !Arrays.asList(sinkColumns).contains(s.getKey()));
    }


    @Override
    public void processBroadcastElement(TableProcess tableProcess, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if ("d".equals(tableProcess.getOp())) {
            broadcastState.remove(tableProcess.getSourceTable());
            configMap.remove(tableProcess.getSourceTable());
        } else {
            broadcastState.put(tableProcess.getSourceTable(),tableProcess);
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }

    }


}
